package com.ehotting.edsta.mqbd.test.http;

import lombok.SneakyThrows;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.admin.OffsetWrapper;
import org.apache.rocketmq.common.admin.TopicOffset;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl;

import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

public class RocketMQExample {

    public static void main(String[] args) throws org.apache.rocketmq.remoting.exception.RemotingException, InterruptedException, org.apache.rocketmq.client.exception.MQClientException {
//        DefaultMQAdminExt adminExt = new DefaultMQAdminExt();
//        adminExt.setNamesrvAddr("192.168.1.30:9876");
//        // 启动服务
//        adminExt.start();
//        TopicList topicList = adminExt.fetchAllTopicList();
//        Set<String> topicList1 = topicList.getTopicList();
//        for (String s : topicList1) {
//            System.out.println(s);
//        }
//        System.out.println("----------topic list------------");
//        // 获取topic: test下的信息
//        TopicRouteData res = adminExt.examineTopicRouteInfo("BlockTopic");
//        System.out.println(res);
        long diffNum = getDiffNumByConsumerGroup();
        //getDiffNumByTopic();
        System.out.println("diffNum:"+diffNum);
    }

    /**
     * 获取消息堆积量
     * @version 1.0
     * @author zhangpeng
     * @return long
     * @exception Exception
     *
     */

    public static long getDiffNumByConsumerGroup(){
        long diffTotal1 = 0L;
        try{
            DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt();
            defaultMQAdminExt.setNamesrvAddr("192.168.1.30:9876");
            defaultMQAdminExt.start();
            //当消费端未消费时，此方法会报错


            //Collections.sort(mqList);
            Runnable r =new Runnable() {
                @SneakyThrows
                @Override
                public void run() {
                    //遍历所有的队列，计算堆积量
                    while (true) {
                        long diffTotal = 0L;
                        ConsumeStats consumeStats = defaultMQAdminExt.examineConsumeStats("test_consumer_group_A","TopicTest");
                        List<MessageQueue> mqList = new LinkedList(consumeStats.getOffsetTable().keySet());
                        for (MessageQueue mq : mqList) {
                            //只计算group下此生产端发送对应的Topic
                            //if ("TopicTest".equals(mq.getTopic())) {
                                OffsetWrapper offsetWrapper = (OffsetWrapper) consumeStats.getOffsetTable().get(mq);
                                long diff = offsetWrapper.getBrokerOffset() - offsetWrapper.getConsumerOffset();
                                diffTotal += diff;
                            //}
                        }
                        System.out.println("getDiffNumByConsumerGroup-diffTotal:" + diffTotal);


                        Thread.sleep(2000);
                    }
                }
            };
            Thread t = new Thread(r);
            t.start();
        }catch(Throwable e){
            e.printStackTrace();
            //logger.error("监控客户端获取消息堆积量异常，未能正常获取消息堆积量，消息堆积量默认设置为0", e);
            //此中出现任何错误，均返回堆积量为0；
            diffTotal1 = 0L;
            /**
             * 此处屏蔽不要了，鉴于这种情况只是刚开始时消费端未消费时会出现，发生频率低，
             * 且捕获异常后获取队列偏移量，无法确定异常类型，
             * 会对后面的逻辑处理造成影响，故不兼容此情况了
             */
            //当只有生产，还未有消费时，上述方法会报错，
            //这是只需获取topic中所有的Queue最大位移和即为消息堆积量
//            try{
//                TopicStatsTable topicStatsTable = defaultMQAdminExt.examineTopicStats(this.topicName);
//                List<MessageQueue> mqList = new LinkedList();
//                mqList.addAll(topicStatsTable.getOffsetTable().keySet());
//                Collections.sort(mqList);
//                diffTotal = 0L;
//                for (MessageQueue mq : mqList) {
//                    TopicOffset topicOffset = (TopicOffset)topicStatsTable.getOffsetTable().get(mq);
//                    long diff = topicOffset.getMaxOffset() - topicOffset.getMinOffset();
//                    diffTotal += diff;
//                }
//            }catch(Throwable e1){
//                logger.error("监控获取消息堆积量出错，返回消息堆积量为0");
//                logger.error(e.getMessage(), e);
//                logger.error(e1.getMessage(), e1);
//                diffTotal = 0L;
//            }
        }
        return diffTotal1;
    }
//
//    public static long getDiffNumByTopic(){
//        long diffTotal1 = 0L;
//        try{
//            DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt();
//            defaultMQAdminExt.setNamesrvAddr("192.168.1.30:9876");
//            defaultMQAdminExt.start();
//            //Collections.sort(mqList);
//            Runnable r =new Runnable() {
//                @SneakyThrows
//                @Override
//                public void run() {
//                    //遍历所有的队列，计算堆积量
//                    while (true) {
//                        //当消费端未消费时，此方法会报错
//                        TopicStatsTable topicStatsTable = defaultMQAdminExt.examineTopicStats(topicName);
//                        List<MessageQueue> mqList = new LinkedList(topicStatsTable.getOffsetTable().keySet());
//                        long diffTotal = 0L;
//                        for (MessageQueue mq : mqList) {
//                            //只计算group下此生产端发送对应的Topic
//                            if (topicName.equals(mq.getTopic())) {
//                                TopicOffset offsetWrapper = (TopicOffset) topicStatsTable.getOffsetTable().get(mq);
//                                long diff = offsetWrapper.getMaxOffset() - offsetWrapper.getMinOffset();
//                                diffTotal += diff;
//                            }
//                        }
//                        System.out.println("Topic-diffTotal:" + diffTotal);
//                        Thread.sleep(2000);
//                    }
//                }
//            };
//            Thread t = new Thread(r);
//            t.start();
//        }catch(Throwable e){
//            e.printStackTrace();
//            //logger.error("监控客户端获取消息堆积量异常，未能正常获取消息堆积量，消息堆积量默认设置为0", e);
//            //此中出现任何错误，均返回堆积量为0；
//            diffTotal1 = 0L;
//            /**
//             * 此处屏蔽不要了，鉴于这种情况只是刚开始时消费端未消费时会出现，发生频率低，
//             * 且捕获异常后获取队列偏移量，无法确定异常类型，
//             * 会对后面的逻辑处理造成影响，故不兼容此情况了
//             */
//            //当只有生产，还未有消费时，上述方法会报错，
//            //这是只需获取topic中所有的Queue最大位移和即为消息堆积量
////            try{
////                TopicStatsTable topicStatsTable = defaultMQAdminExt.examineTopicStats(this.topicName);
////                List<MessageQueue> mqList = new LinkedList();
////                mqList.addAll(topicStatsTable.getOffsetTable().keySet());
////                Collections.sort(mqList);
////                diffTotal = 0L;
////                for (MessageQueue mq : mqList) {
////                    TopicOffset topicOffset = (TopicOffset)topicStatsTable.getOffsetTable().get(mq);
////                    long diff = topicOffset.getMaxOffset() - topicOffset.getMinOffset();
////                    diffTotal += diff;
////                }
////            }catch(Throwable e1){
////                logger.error("监控获取消息堆积量出错，返回消息堆积量为0");
////                logger.error(e.getMessage(), e);
////                logger.error(e1.getMessage(), e1);
////                diffTotal = 0L;
////            }
//        }
//        return diffTotal1;
//    }


}